#include <stdio.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h> // For atoi function
#include <pthread.h> // For pthreads

#include "mosquitto.h"
#include "cJSON.h"
#include "linux/can.h"
#include "can_rx.h" 
#include "can_tx.h"
#include "modbus_poll.h"

typedef struct s_conn_para {
	char *hostname;
	char *pub_topic;
	char *clientId;
	char *username;
	char *password;
	int port;
	char payload[256];
	char *sub_topic;
	int keepalive;
	int qos;
	pthread_mutex_t mqtt_mutex; // Mutex for MQTT operations
} conn_para_t;		//MQTT connection parameters

void mqtt_clean(struct mosquitto *mosq);  //MQTT Clean
static void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg); //parse subscription message formats

void *mqtt_subscribe(void *arg);  //mqtt subscribe
void *mqtt_publish(void *arg);    //mqtt publish

conn_para_t conn_para = {
	"iot-06z00i9k03hcz6s.mqtt.iothub.aliyuncs.com",
	"/sys/h098jXhNP9d/mqtt/thing/event/property/post",
	"h098jXhNP9d.mqtt|securemode=2,signmethod=hmacsha256,timestamp=1721637242760|",
	"mqtt&h098jXhNP9d",
	"75a00f16f3102fbfb8a211a0dcd21dd9548c62b418b44478a04bd57c1f74b6bd",	
	1883,
	"",
	"/sys/h098jXhNP9d/mqtt/thing/event/property/post_reply",
	60,
	0,
	PTHREAD_MUTEX_INITIALIZER
};

int main(int argc, char **argv)
{
	struct mosquitto *mosq;		//Define the mqtt handle
	pthread_t sub_thread, pub_thread;  //Define the pthread handle
	int ret;

	const char *interface_name = "can0";	//Specify can interface


	mosquitto_lib_init();  //mosquitto initialize

	mosq = mosquitto_new(conn_para.clientId, true, NULL); //create mosquitto handle

	if (mosquitto_username_pw_set(mosq, conn_para.username, conn_para.password) != MOSQ_ERR_SUCCESS)//Set connection parameters
       	{
		printf("set connection parametes failure: %s\n", strerror(errno));
		goto cleanup;
	}

	printf("mqtt connect parameters set OK\n");

	if (mosquitto_connect(mosq, conn_para.hostname, conn_para.port, conn_para.keepalive) != MOSQ_ERR_SUCCESS)
	//mosquitto connect function
	{
		printf("connect server failure: %s\n", strerror(errno));
		goto cleanup;
	}
	printf("connect ok\n");

	mosquitto_message_callback_set(mosq, my_message_callback);	//Set callback function

	// Create thread for subscribing
	ret = pthread_create(&sub_thread, NULL, mqtt_subscribe, (void *)mosq);  
	if (ret != 0) {
		printf("Failed to create subscribe thread: %s\n", strerror(ret));
		goto cleanup;
	}

	// Create thread for publishing
	ret = pthread_create(&pub_thread, NULL, mqtt_publish, (void *)mosq);
	if (ret != 0) { 
		printf("Failed to create publish thread: %s\n", strerror(ret));
		goto cleanup;
	}

	// Join threads (wait for them to finish)
	pthread_join(sub_thread, NULL);
	pthread_join(pub_thread, NULL);

cleanup:
	mqtt_clean(mosq);
	return 0;
}


void mqtt_clean(struct mosquitto *mosq)
{
	mosquitto_destroy(mosq);
	mosquitto_lib_cleanup();
}

static void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{
	//    printf("Received message: %s\n", (char *)msg->payload);   
	int can_fd = send_can_message("can0"); //Specify can interface
	char  buf[200] = {0};
	struct can_frame frame;			//can Message struct

	cJSON *root = cJSON_Parse((char *)msg->payload); //Para the cJson payload in MQTT message 
	if (root == NULL)
	{
		printf("JSON parsing error\n");
		return;
	}

	cJSON *method = cJSON_GetObjectItem(root, "method");  //Get the "method" field from the JSON root object
	if (method != NULL && cJSON_IsString(method))  //if the "method" field exists and is of string type
	{
		if (strcmp(method->valuestring, "thing.service.property.set") == 0)   //
		{
			cJSON *params = cJSON_GetObjectItem(root, "params");
			if (params != NULL)
			{
				cJSON *can_id = cJSON_GetObjectItem(params, "can_id");
				cJSON *can_dlc = cJSON_GetObjectItem(params, "can_dlc");
				cJSON *can_data0 = cJSON_GetObjectItem(params, "can_data0");
				cJSON *can_data1 = cJSON_GetObjectItem(params, "can_data1");
				cJSON *can_data2 = cJSON_GetObjectItem(params, "can_data2");
				cJSON *can_data3 = cJSON_GetObjectItem(params, "can_data3");
				cJSON *can_data4 = cJSON_GetObjectItem(params, "can_data4");
				cJSON *can_data5 = cJSON_GetObjectItem(params, "can_data5");
				cJSON *can_data6 = cJSON_GetObjectItem(params, "can_data6");
				cJSON *can_data7 = cJSON_GetObjectItem(params, "can_data7");

				if (cJSON_IsNumber(can_id) && cJSON_IsNumber(can_dlc))
				{
					int id = can_id->valueint;
					int dlc = can_dlc->valueint;
					frame.can_id = id;
					frame.can_dlc = dlc;

					// Make sure dlc is within valid range (0 to 8 as per CAN standard)
					if (dlc >= 0 && dlc <= 8)
					{
						// Array to store can data
						double can_data[8] = {0}; // Initialize with zeros

						// Read can data into the array
						cJSON *can_data_items[8] = {can_data0, can_data1, can_data2, can_data3,
							can_data4, can_data5, can_data6, can_data7};

						for (int i = 0; i < dlc; ++i)
						{
							if (cJSON_IsNumber(can_data_items[i])) //Check whether the type is an integer
							{
								can_data[i] = can_data_items[i]->valuedouble;
								frame.data[i] = can_data_items[i]->valuedouble;
							}
						}

						// Print received CAN message
						printf("Receive CAN message: id=%d, dlc=%d, data=", id, dlc);
						int index = snprintf(buf,sizeof(buf),"CAN message: id=%d, dlc=%d ",id, dlc);
						for (int i = 0; i < dlc; i++)
						{
							printf("%.2f ", can_data[i]);
							index += snprintf(buf+index,sizeof(buf),"data[%d]: %.2f",i,can_data[i]); 
						}
						printf("\n");
						if(write(can_fd, &frame, sizeof(struct can_frame))!= sizeof(struct can_frame))
						{
							printf("write failure: %s\n", strerror(errno));
							close(can_fd);
						}
						printf("write %s\n", buf);
						memset(buf,0, sizeof(buf));
					}
					//memset(buf, 0, sizeof(buf));
					else
					{
						printf("Invalid DLC value: %d\n", dlc);
					}
				}
				else
				{
					printf("Invalid or missing CAN data fields\n");
				}
			}
			else
			{
				printf("Missing params object in message\n");
			}
		}
		else
		{
			//    printf("Unsupported method: %s\n", method->valuestring);
		}
	}
	else
	{
		printf("Missing or invalid method field\n");
	}

	cJSON_Delete(root);
}



void *mqtt_subscribe(void *arg)
{
	struct mosquitto *mosq = (struct mosquitto *)arg;  
	int mid;
	conn_para_t *conn = &conn_para;

	if (mosquitto_subscribe(mosq, &mid, conn->sub_topic, conn->qos) != MOSQ_ERR_SUCCESS) {
		printf("mosquitto subscribe failure: %s\n", strerror(errno));
		return NULL;
	}
	printf("subscribe OK\n");

	while (1) {
		if (mosquitto_loop(mosq, -1, 1) != MOSQ_ERR_SUCCESS) {
			printf("Mosquitto loop error\n");
			break;
		}
	}

	return NULL;
}

void *mqtt_publish(void *arg)
{
	struct 		mosquitto *mosq = (struct mosquitto *)arg;
	modbus_t 	*ctx;
	int 		mid1;
	conn_para_t 	*conn = &conn_para;
	int 		modbus_connect_flag = 0;
	int		rc = 0;
	uint16_t	tab_reg[32];

	while (1) {

		if(modbus_connect_flag == 0)
		{
			if( (ctx = modbus_new_rtu("/dev/ttymxc3", 115200, 'N', 8, 1)) == NULL)
			{
				printf("modbus_new_rtu failure\n");
			}	
			modbus_set_slave(ctx, 1);//从机端设置从机地址
			if(modbus_connect(ctx) == -1)
			{
				printf("modbus_connect failure: %s\n", modbus_strerror(errno));
				//modbus_connect_flag = 0;
			}
			else
			{
				modbus_connect_flag = 1;
			}
		}

		rc = modbus_read_registers(ctx, 0, 2, tab_reg);
		if(rc == -1)
		{
			printf("modbus read failure: %s\n", modbus_strerror(errno));
			modbus_connect_flag = 0;
			modbus_close(ctx);
			modbus_free(ctx);
		}

		struct can_frame received_frame = receive_can_message("can0"); //
		int index = 0;

		// 构建 JSON payload
		index += snprintf(conn->payload + index, sizeof(conn->payload) - index,
				"{\"method\":\"thing.service.property.set\",\"id\":\"10353468\",\"params\":{\"can_id\":%d,\"can_dlc\":%d,",
				(int)received_frame.can_id, (int)received_frame.can_dlc);

		// 添加 can_data 字段到 params
		for (int i = 0; i < received_frame.can_dlc; i++) {
			//if (i == received_frame.can_dlc - 1) {
			//	index += snprintf(conn->payload + index, sizeof(conn->payload) - index,
			//			"\"can_data%d\":%d", i, (int)received_frame.data[i]);
			//} else {
				index += snprintf(conn->payload + index, sizeof(conn->payload) - index,
						"\"can_data%d\":%d,", i, (int)received_frame.data[i]);
			//}
		}

		for(int i=0; i<rc; i++)
		{
			if( i==rc-1 )
			{
				index += snprintf(conn->payload + index, sizeof(conn->payload)-index, "\"modbus_noisy\":%d",tab_reg[i]);
			}
			else
			{
				index += snprintf(conn->payload + index, sizeof(conn->payload)-index, "\"modbus_lux\":%d,",tab_reg[i]);

			}
		}
		// add the closing brachet and version field
		index += snprintf(conn->payload + index, sizeof(conn->payload) - index,
				"},\"version\":\"1.0.0\"}");

		// Print the generated JSON payload for debugging(option)
		//printf("生成的 JSON payload: %s\n", conn->payload);
		pthread_mutex_lock(&conn->mqtt_mutex);
		int ret = mosquitto_publish(mosq, &mid1, conn->pub_topic, strlen(conn->payload), conn->payload, conn->qos, 0);
		pthread_mutex_unlock(&conn->mqtt_mutex);

		if (ret != MOSQ_ERR_SUCCESS) {
			printf("mosquitto publish data failure: %s\n", strerror(errno));
			continue;
		}
		printf("published message: %s\n", conn->payload);
		sleep(1);
	}

	return NULL;
}

